# hive -f  文件路径
# hive -hivevar input_id=3 -f 文件路径
# where etl_date = ${hivevar:input_id}
# ${hivevar:input_id}
#


# 先编写SQL文件
# 区分哪些表是全量覆盖，哪些表是全量快照，哪些表是增量抽取
# 一张一个SQL文件，文件名：dwd_inc_表名.SQL
# SQL 预留变量
# 编写 sh 脚本
# 调用 9 个增量脚本文件。
# 提示：hive -hiveconf  -f
# 后台运行，并记录日志到 /aml/dwd/表名_YYYYMMDD.log
# 通过 crontab 实现每天凌晨 一点半 执行增量脚本


#!/bin/bash

dt=`date -d '-1 day' +"%Y-%m-%d"`
dtt=`date +"%Y%m%d"`


#dwd
echo "rang dwd_inc_dim_aml_customer.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_dim_aml_customer.sql > /aml/dwd/dim_aml_customer_"$dtt.log" &

echo "rang dwd_inc_dim_aml_account.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_dim_aml_account.sql > /aml/dwd/dim_aml_account_"$dtt.log" &

echo "rang dwd_inc_dim_aml_watchlist.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_dim_aml_watchlist.sql >/aml/dwd/dim_aml_watchlist_"$dtt.log" &

echo "rang dwd_inc_dim_aml_rule.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_dim_aml_rule.sql > /aml/dwd/dim_aml_rule_"$dtt.log" &

echo "rang dwd_inc_fact_aml_transaction.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_fact_aml_transaction.sql > /aml/dwd/fact_aml_transaction_"$dtt.log" &

echo "rang dwd_inc_fact_aml_screening.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_fact_aml_screening.sql > /aml/dwd/fact_aml_screening_"$dtt.log" &

echo "rang dwd_inc_fact_aml_alert.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_fact_aml_alert.sql > /aml/dwd/fact_aml_alert_"$dtt.log" &

echo "rang dwd_inc_fact_aml_str_report.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dwd_inc_fact_aml_str_report.sql >/aml/dwd/fact_aml_str_report_"$dtt.log" &

# dws
echo "rang dws_dws_aml_risk_dashboard.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dws_dws_aml_risk_dashboard.sql > /aml/dwd/risk_dashboard_"$dtt.log" &

echo "rang dws_aml_transaction_behavior.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dws_aml_transaction_behavior.sql > /aml/dwd/transaction_behavior_"$dtt.log" &

echo "rang dws_aml_customer_risk_profile.sql"
nohup hive -hivevar dt=$dt  -f /root/dwd/dws_aml_customer_risk_profile.sql >/aml/dwd/customer_risk_profile_"$dtt.log" &



rm -rf `find /root/aml/ -mtime 2`

crontab  "*分 *时 *日 *月 *星 /命令"

##############################################################################
 --1.客户维度  dwd_inc_dim_aml_customer.sql  /aml/dwd/dim_aml_customer_"$dtt.log"
insert into dwd.dim_aml_customer partition (etl_date = '2015-08-15')
select customer_id,
       cust_no,
       cust_name,
       case when cust_type='IND' then '个人'
            when cust_type='CORP' then '对公'
            else cust_type
        end as cust_type,
       case when id_type = 'ID_CARD' then '身份证'
            when id_type = 'PASSPORT' then '护照'
            when id_type = 'BUS_LIC' then '营业执照'
            else id_type
        end as id_type,
       case when risk_level = 'LOW' then '低'
            when risk_level = 'MEDIUM' then '中'
            when risk_level = 'HIGH' then '高'
            when risk_level = 'VERY_HIGH' then '极高'
            else risk_level
        end as  risk_level,
       case when risk_level_source = 'MANUAL' then '人工'
            when risk_level_source = 'AUTO' then '系统自动'
            else risk_level_source
        end as risk_level_source,
       case when status ='ACTIVE' then '活跃'
            when status ='INACTIVE' then '休眠'
            when status ='CLOSED' then '已销户'
            when status ='FROZEN' then '冻结'
            else status
        end as status,
       industry as industry_type ,
       occupation as occupation_type  ,
       nationality as country_code  ,
       residence_city as city_code ,
       open_date as start_date ,
       '9999-12-31'  as end_date ,
       true is_current
from ods.ods_aml_customer_master
where close_date is null;


--2,账户维度表  dwd_inc_dim_aml_account.sql   /aml/dwd/dim_aml_account_"$dtt.log"
insert into dwd.dim_aml_account partition (etl_date = '${hivevar:dt}')
select account_id,
       customer_id,
       acct_no  ,
       case when acct_type ='SAVING' then '储蓄'
            when acct_type ='LOAN' then '贷款'
            when acct_type ='INVESTMENT' then '投资'
            when acct_type ='CREDIT' then '信用卡'
            when acct_type ='CURRENT' then '活期'
            else acct_type
        end as acct_type,
       CURRENCY as currency_code,
       current_balance  ,
       avg_daily_balance,
       case when status ='ACTIVE' then '活跃'
            when status ='INACTIVE' then '休眠'
            when status ='CLOSED' then '已销户'
            when status ='FROZEN' then '冻结'
            else status
        end as status,
       branch_code      ,
       case when channel_open ='COUNTER' then '柜面'
            when channel_open ='ONLINE' then '网银'
            when channel_open ='MOBILE' then '手机银行'
            else channel_open
        end as channel_open,
       open_date as start_date ,
       '9999-12-31' as end_date
from ods.ods_aml_account_master
where etl_date = '${hivevar:dt}';

--3. 名单表 dwd_inc_dim_aml_watchlist.sql   /aml/dwd/dim_aml_watchlist_"$dtt.log"
insert into dwd.dim_aml_watchlist PARTITION (etl_date = '${hivevar:dt}')
select d.entity_id as entity_id,
       m.list_code as list_code,
       case
           when m.list_type = 'TERRORIST' then '恐怖'
           when m.list_type = 'CRIMINAL' then '犯罪'
           when m.list_type = 'SANCTIONS' then '制裁'
           when m.list_type = 'PEP' then '政要'
           else '其他'
        end as status,
       case
           when m.list_source = 'EU' then '欧盟'
           when m.list_source = 'UN' then '联合国'
           when m.list_source = 'HKMA' then '香港'
           when m.list_source = 'OFAC' then '美国财政部'
           else '内部'
        end as list_source,
       case
           when d.entity_type = 'PERSON' then '个人'
           else '组织'
        end as entity_type,
       case
           when d.sanction_type = 'ASSET_FREEZE' then '资产冻结'
           when d.sanction_type = 'TRAVEL_BAN' then '旅行禁令'
           when d.sanction_type = 'TRADE_EMBARGO' then '贸易禁运'
           else '金融制裁'
        end as sanction_type
from ods.ods_aml_watchlist_detail d
left join ods.ods_aml_watchlist_master m on d.list_id = m.list_id
where etl_date = '${hivevar:dt}';

--4.维度表 规则表 dwd_inc_dim_aml_rule.sql   /aml/dwd/dim_aml_rule_"$dtt.log"
insert overwrite table dwd.dim_aml_rule partition (etl_date = '2025-08-15')
select rule_id,
       rule_name,
       case when rule_type = 'AMOUNT' then "金额"
            when rule_type = 'FREQUENCY' then "频率"
            when rule_type = 'PATTERN' then "模式"
            when rule_type = 'BEHAVIOR' then "行为"
        end as rule_type,
       case when rule_category = 'CASH' then "现金"
            when rule_category = 'TFR' then "转账"
            when rule_category = 'CROSS_BORDER' then "跨境"
            when rule_category = 'GAMBLING' then "赌博"
            when rule_category = 'INVESTMENT' then "投资"
            when rule_category = 'CREDIT' then "信用"
            when rule_category = 'LOAN' then "贷款"
        end as rule_category,
       case when severity_level = 'LOW' then "低"
            when severity_level = 'MEDIUM' then "中"
            when severity_level = 'HIGH' then "高"
            when severity_level = 'CRITICAL' then "严重"
        end as severity_level,
       case when active = 'Y' then "是"
            when active = 'N' then "否"
        end as active_status,
       nvl(get_json_object(param_json,"$.threshold"),0) as threshold
from ods.ods_aml_monitoring_rule;

--5. 交易事实表 dwd_inc_fact_aml_transaction.sql   /aml/dwd/fact_aml_transaction_"$dtt.log"
insert into dwd.fact_aml_transaction partition (etl_date = '${hivevar:dt}')
select cust.customer_id as customer_sk,
       acc.account_id   as account_sk,
       case
           when acc.currency = 'CNY' then 1
           when acc.currency = 'EUR' then 2
           when acc.currency = 'HKD' then 3
           when acc.currency = 'JPY' then 4
           when acc.currency = 'USD' then 5
        end as currency_sk,
       --CNY 1,EUR 2,HKD 3,JPY 4,USD 5
       case
           when txn_type = 'CASH_DEP' then '现金存入'
           when txn_type = 'CASH_WD' then '现金取款'
           when txn_type = 'TFR_IN' then '转入'
           when txn_type = 'TFR_OUT' then '转出'
           when txn_type = 'PAYMENT' then '支付'
           else txn_type
        end as txn_type,
       txn_sub_type,
       case
           when txn_channel = 'COUNTER' then '柜面'
           when txn_channel = 'ATM' then 'ATM'
           when txn_channel = 'ONLINE' then '网银'
           when txn_channel = 'MOBILE' then '手机银行'
           when txn_channel = 'POS' then 'POS'
           else txn_channel
        end as txn_channel,
       case
           when txn_status = 'SUCCESS' then '成功'
           when txn_status = 'FAILED' then '失败'
           when txn_status = 'PENDING' then '待处理'
           else txn_status
        end as txn_status,
       txn_amount as amount,
       case
           when remitter_country is null or beneficiary_country is null then true
           when remitter_country != beneficiary_country then true
           when remitter_country = beneficiary_country then false
        end as is_cross_border,
       case
           when risk_level = 'HIGH' or risk_level = 'VERY_HIGH' then true
           else false
        end as is_high_risk
from ods.ods_aml_customer_master cust
join ods.ods_aml_account_master acc
    on cust.customer_id = acc.customer_id
join ods.ods_aml_transaction_detail tran
    on acc.account_id = tran.account_id
where etl_date = '${hivevar:dt}';

--6. 筛查事实表 dwd_inc_fact_aml_screening.sql   /aml/dwd/fact_aml_screening_"$dtt.log"
insert into dwd.fact_aml_screening partition (etl_date = '${hivevar:dt}')
select c.customer_id as customer_sk,
       wd.list_id as watchlist_sk,
       case
           when s.screening_type = 'CUSTOMER' THEN '客户开户'
           when s.screening_type = 'PERIODIC' THEN '交易'
           else '定期回溯'
        end as screening_type,
       case
           when s.target_type = 'CUSTOMER' THEN '客户'
           when s.target_type = 'ACCOUT' THEN '账户'
           else '交易对手'
        end as target_type,
       case
           when s.match_level = 'EXACT' THEN '完全匹配'
           when s.match_level = 'FUZZY_HIGH' THEN '模糊高'
           when s.match_level = 'FUZZY_MED' THEN '模糊中'
           when s.match_level = 'FUZZY_LOW' THEN '模糊低'
           else s.match_level
        end as match_level,
       case
           when s.screening_status = 'PENDING' THEN '待审'
           when s.screening_status = 'ALERT' THEN '警报'
           when s.screening_status = 'FALSE_POSITIVE' THEN '误报'
           when s.screening_status = 'CONFIRMED' THEN '确认'
           else s.screening_status
        end as screening_status,
       s.match_score as match_score
from ods.ods_aml_screening_result s
left join ods.ods_aml_watchlist_master ws
        on ws.list_id = s.matched_list_id
left join ods.ods_aml_watchlist_detail wd
        on wd.list_id = ws.list_id
left join ods.ods_aml_customer_master c
        on c.customer_id = s.target_id
where etl_date = '${hivevar:dt}';


--7预警事实表  dwd_inc_fact_aml_alert.sql   /aml/dwd/fact_aml_alert_"$dtt.log"
insert into dwd.fact_aml_alert partition (etl_date = '${hivevar:dt}')
select aal.customer_id as customer_sk,
       aal.account_id as account_sk,
       aal.rule_id as rule_sk,
       aal.alert_date date_sk,
        -- 预警类型 (rule:规则, model:模型)
       case when aal.alert_type = 'RULE' then "规则"
            when aal.alert_type = 'MODEL' then "模型"
           else aal.alert_type
        end as alert_type,
    -- 预警状态 (PENDING:待处理, REVIEWING:审核中, DISMISSED:排除, CONFIRMED:确认)
       case when aal.alert_status = 'PENDING' then "待处理"
            when aal.alert_status = 'REVIEWING' then "审核中"
            when aal.alert_status = 'DISMISSED' then "排除"
            when aal.alert_status = 'CONFIRMED' then "确认"
           else aal.alert_status
        end as alert_status,
    -- 严重等级 (low, medium, high, critical)
       case when aal.severity_level = 'low' then "低"
            when aal.severity_level = 'medium' then "中"
            when aal.severity_level = 'high' then "高"
            when aal.severity_level = 'critical' then "危急"
           else aal.severity_level
        end as severity_level,
       trigger_amount,
       trigger_txn_count
from ods.ods_aml_alert aal
left join dwd.dim_aml_customer acu
    on aal.customer_id = acu.customer_id
left join dwd.dim_aml_account aac
    on aac.account_id = aal.account_id
left join dwd.dim_aml_rule aru
    on aru.rule_id = aal.rule_id
left join dwd.dim_aml_date ada
    on ada.full_date = aal.alert_date
where etl_date = '${hivevar:dt}';




--8报告事实表  dwd_inc_fact_aml_str_report.sql   /aml/dwd/fact_aml_str_report_"$dtt.log"
insert into dwd.fact_aml_str_report partition (etl_date = '${hivevar:dt}')
select atr.customer_id as customer_sk,
       atr.report_date as date_sk,
       atr.first_txn_date as first_txn_date_sk,
       atr.last_txn_date as last_txn_date_sk,
       -- 报告类型 (INITIAL:初始, AMENDMENT:修正, CLOSURE:结案)
       case when atr.report_type = 'INITIAL' then "初始"
            when atr.report_type = 'AMENDMENT' then "修正"
            when atr.report_type = 'CLOSURE' then "结案"
           else atr.report_type
        end as report_type,
       --案件类别 (MONEY_LAUNDERING:洗钱, TERRORIST_FINANCING:恐怖融资, FRAUD:欺诈等)
       case when atr.case_category = 'MONEY_LAUNDERING' then "洗钱"
            when atr.case_category = 'TERRORIST_FINANCING' then "恐怖融资"
            when atr.case_category = 'FRAUD' then "欺诈"
            when atr.case_category = 'CORRUPTION' then "腐败"
            when atr.case_category = 'TAX_EVASION' then "逃税"
           else atr.case_category
        end as case_category,
       -- 报告状态 (DRAFT:草稿, SUBMITTED:已提交, WITHDRAWN:已撤回)
       case when report_status = 'DRAFT' then "草稿"
           when report_status = 'SUBMITTED' then "已提交"
           when report_status = 'WITHDRAWN' then "已撤回"
           else atr.report_status
        end as report_status,
       total_amount
from ods.ods_aml_suspicious_txn_report atr
left join dwd.dim_aml_customer acu
        on atr.customer_id = acu.customer_id
where etl_date ='${hivevar:dt}';

-- 客户风险画像宽表

with balance as (
    select c.customer_id
         , sum(a.current_balance) as total_balance
    from dwd.dim_aml_customer c
             join dwd.dim_aml_account a
                  on c.customer_id = a.customer_id
    group by c.customer_id
), last_jiaoyi as(
    select
        c.customer_id
        ,max(f.date_sk) as last_txn_date
    from dwd.dim_aml_customer c
    join dwd.dim_aml_account a
      on c.customer_id = a.customer_id
    join dwd.fact_aml_transaction f
      on a.account_id = f.account_sk
    group by c.customer_id
),alert as (
    select
        a.customer_sk
        ,count(*) as alert_count
    from dwd.fact_aml_alert a
    group by a.customer_sk
),screening as (
    select
        s.customer_sk
        ,count(*) as screening_hit_count
    from dwd.fact_aml_screening s
    group by s.customer_sk
)
insert into dws.dws_aml_customer_risk_profile partition (etl_date = '${hivevar:dt}')
select
     c.customer_id
    ,c.risk_level
    ,b.total_balance
    ,l.last_txn_date
    ,coalesce(al.alert_count,0) as alert_count
    ,coalesce(s.screening_hit_count,0) as screening_hit_count
    ,coalesce(r.report_status,'false') as str_submitted
    ,case when c.industry_type in ('金融','贸易','珠宝','房地产') then true else false end industry_risk_flag
 from dwd.dim_aml_customer c
left join balance b
    on c.customer_id = b.customer_id
left join last_jiaoyi l
    on l.customer_id = c.customer_id
left join alert al
    on al.customer_sk = c.customer_id
left join screening s
    on s.customer_sk = c.customer_id
left join dwd.fact_aml_str_report r
    on r.customer_sk = c.customer_id
where date_sk = '${hivevar:dt}';


-- 风险监控
insert into dws.dws_aml_risk_dashboard
partition (etl_date = '${hivevar:dt}' )
select
    c.start_date as date_sk
    ,count(distinct c.customer_id) as new_customer_count
    ,count(case when c.risk_level in ('高','极高') then 1 else null end) as high_risk_customer_count
    ,count(a.trigger_txn_count) as alert_count
    ,count(case when a.alert_status = '确认' then 1 else null end) as confirmed_alert_count
    ,count(case when r.report_status = '已提交' then 1 else null end) as str_submitted_count
    ,sum(case when t.is_cross_border then 1 else 0 end)  as cross_border_txn_count
    ,count(t.is_cross_border) as total_txn_count
from dwd.dim_aml_customer c
left join dwd.fact_aml_alert a
on c.customer_id = a.customer_sk
left join dwd.fact_aml_str_report r
on r.customer_sk = a.customer_sk
left join dwd.fact_aml_transaction t
on t.customer_sk = r.customer_sk
where date_sk = '${hivevar:dt}'
group by c.start_date;




-- 交易行为分析
insert overwrite table dws.dws_aml_transaction_behavior partition (etl_date = '${hivevar:dt}')
select
    t.date_sk
    ,t.txn_type
    ,t.txn_channel
    ,t.is_high_risk
    ,count(*) as txn_count
    ,sum(amount) as total_amount
    ,avg(amount) as avg_amount
    ,round(sum(case when t.is_cross_border then 1 else 0 end) / count(*) * 100,2) as cross_border_ratio
from dwd.fact_aml_transaction t
where date_sk = '${hivevar:dt}'
group by
    t.date_sk
    ,t.txn_type
    ,t.txn_channel
    ,t.is_high_risk;





